#include "QueueManager.h"
#include <cstdint>
#include <cstring>
#include <memory>
#include <set>
#include <sstream>
#include <unistd.h>
#include "ConfMgr.h"
#include "FIFOQueue.h"
#include "MessageHeader.h"
#include "PlatLogger.h"
#include "Util.h"
#include "XnccType.h"
namespace xncc {
namespace foundation {
    using namespace types;
    using msg::FIFOQueue;
    using msg::MsgHeaderInThread;

    static constexpr uint32_t THREAD_MSG_HEAD_LEN = sizeof(MsgHeaderInThread);
    static constexpr uint32_t IPC_MSG_HEAD_LEN    = sizeof(MsgHeaderInThread) + sizeof(Header);

    QueueManager::~QueueManager() = default;
    int QueueManager::init(const config::ConfMgr& cfg)
    {
        const auto& ioQueues = cfg.getIPCPubQueues();

        ska::flat_hash_map<thread_id_t, std::set<queue_id_t>> threadPubIds;
        // 由于一个IPC线程只有pubid,所以可以拿线程ID做key
        for (const auto& ele : ioQueues) {
            const auto ipcThreadId = static_cast<thread_id_t>(ele.first);
            for (const auto& ptr : ele.second) {
                auto       queuePtr = std::make_shared<FIFOQueue>(ptr->Size, ptr->AutoExpand);
                const auto subTid   = static_cast<thread_id_t>(ptr->SubTid);
                const auto realId   = queuePtr->id();
                TRACE_INFO << "IPC thread:" << ipcThreadId << " will send message to business thread " << subTid
                           << " by queue id:" << ptr->Id << "(real id:" << realId << ")";
                ioThreadPubQueues_[ipcThreadId].push_back(queuePtr);
                // Queue只能被业务线程订阅
                busiThreadSubTcpQueues_[subTid].push_back(queuePtr);

                threadPubIds[ipcThreadId].insert(ptr->Id);
                pubidQueues_[ptr->Id].push_back(queuePtr);
            }
        }
        // 剩下的是业务线程之间的订阅关系了
        const auto& busiQueues = cfg.getBusiPubQueues();
        for (const auto& ele : busiQueues) {
            const auto busiThreadId = static_cast<thread_id_t>(ele.first);
            for (const auto& ptr : ele.second) {
                auto queuePtr = std::make_shared<FIFOQueue>(ptr->Size, ptr->AutoExpand);
                threadPubIds[busiThreadId].insert(ptr->Id);
                pubidQueues_[ptr->Id].push_back(queuePtr);
                // Queue只能被业务线程订阅
                const auto subTid = static_cast<thread_id_t>(ptr->SubTid);
                const auto realId = queuePtr->id();
                busiThreadSubBusiQueues_[subTid].push_back(queuePtr);
                std::ostringstream oss;
                for (auto msgCode : ptr->MsgCodes) {
                    msgCodePubQueues_[static_cast<message_id_t>(msgCode)].push_back(queuePtr);
                    oss << msgCode << ",";
                }
                TRACE_INFO << "business Thread:" << busiThreadId << " will send message (" << oss.str()
                           << ") to business thread " << subTid << " by queue id:" << ptr->Id << "(real id:" << realId
                           << ")";
            }
        }

        for (auto& ele : threadPubIds) {
            for (auto pubid : ele.second) {
                threadPubIds_[ele.first].push_back(pubid);
            }
        }
        return 0;
    }

    FIFOQueue::s_ptr QueueManager::subQueue(queue_id_t queueId, queue_size_t queueSize, bool autoExpand)
    {
        /// 如果在代码中自己启动了一个线程，来发数据的话，那么就需要动态的添加一个Queue 来使用
        /// 这个函数必须接收方来调用
        /// 动态添加的必须是一对一的，不允许接收方创建一个， 发送方多个线程都使用这一个
        auto iter = DynamicBidBuffers_.find(queueId);
        if (iter == DynamicBidBuffers_.end()) {
            auto ptr                    = std::make_shared<FIFOQueue>(queueSize, autoExpand);
            auto dyPtr                  = std::make_shared<DynamicFIFOBuffer>();
            dyPtr->QueueWrapper         = ptr;
            DynamicBidBuffers_[queueId] = dyPtr;
            return ptr;
        }
        return nullptr;
    }

    FIFOQueue::s_ptr QueueManager::createIPCThreadRecvQueue(thread_id_t ioId, thread_id_t busiId)
    {
        auto ptr = std::make_shared<FIFOQueue>(queue_size_t_limit<queue_size_t>::dft(), true);
        ptr->setNeedNotify();  // IPC线程 队列需要通知
        IPCRecvBuffers_[ioId][busiId] = ptr;
        return ptr;
    }

    int QueueManager::pubMsgToIPCThread(const char*     buf,
                                        uint32_t        len,
                                        thread_id_t     ioId,
                                        thread_id_t     busiId,
                                        connection_id_t connId,
                                        message_id_t    ioMsgId,
                                        message_id_t    id)
    {
        auto iter = IPCRecvBuffers_.find(ioId);
        if (iter != IPCRecvBuffers_.end()) {
            auto iter2 = iter->second.find(busiId);
            if (iter2 != iter->second.end()) {
                return sendMsgToIPCThread(iter2->second, buf, len, id, connId, static_cast<thread_id_t>(ioMsgId));
            }
        }
        return -1;
    }

    int QueueManager::sendMsgToIPCThread(const msg::FIFOQueue::s_ptr& ptr,
                                         const char*                  buf,
                                         uint32_t                     len,
                                         message_id_t                 id,
                                         connection_id_t              connId,
                                         thread_id_t                  tid)

    {
        // 复用tid为 msgid
        char* writeBuf = ptr->getWriteBuffer(static_cast<queue_size_t>(IPC_MSG_HEAD_LEN + len));
        if (writeBuf) {
            auto* hptr    = reinterpret_cast<MsgHeaderInThread*>(writeBuf);
            hptr->len     = len + static_cast<uint32_t>(sizeof(Header));
            hptr->id      = id;
            hptr->connId  = connId;
            hptr->tid     = tid;
            Header header = msg::MessageHeader::encode(static_cast<message_id_t>(tid), len);
            memcpy(writeBuf + THREAD_MSG_HEAD_LEN, reinterpret_cast<char*>(&header), sizeof(Header));
            memcpy(writeBuf + IPC_MSG_HEAD_LEN, buf, static_cast<uint32_t>(len));
            ptr->finishWrite();
            return 0;
        }
        else {
            return -1;
        }
    }

    int QueueManager::sendMsgInThread(const msg::FIFOQueue::s_ptr& ptr,
                                      const char*                  buf,
                                      uint32_t                     len,
                                      message_id_t                 id,
                                      connection_id_t              connId,
                                      thread_id_t                  tid)
    {  /// 线程间发送只需要设置msg即可
        char* writeBuf = ptr->getWriteBuffer(static_cast<queue_size_t>(THREAD_MSG_HEAD_LEN + len));
        if (writeBuf != nullptr) {
            auto* hptr   = reinterpret_cast<MsgHeaderInThread*>(writeBuf);
            hptr->len    = len;
            hptr->id     = id;
            hptr->connId = connId;
            hptr->tid    = tid;
            memcpy(writeBuf + THREAD_MSG_HEAD_LEN, buf, static_cast<uint32_t>(len));
            ptr->finishWrite();
            return 0;
        }
        return -1;
    }

    /// 发送者发送动态的Queue数据
    //  这个是线程间的数据发送
    int QueueManager::dynamicPubTreadData(const char* buf, uint32_t len, message_id_t id, queue_id_t bid)
    {
        auto iter = DynamicBidBuffers_.find(bid);
        if (iter == DynamicBidBuffers_.end()) {
            return 1;
        }
        auto ptr = iter->second->QueueWrapper;
        TRACE_DEBUG << ("another Thread send message:") << id << " to queue id:" << ptr->id();
        return sendMsgInThread(ptr, buf, len, id);
    }

    int QueueManager::dynamicPubTreadDataWithLock(const char* buf, uint32_t len, message_id_t id, queue_id_t queueId)
    {  // 日志线程会调用此函数，因此函数内部不能写日志，否则会造成死锁
        // 如果业务线程要接收第三方线程的数据
        // 但是第三方的回调 函数并不是一个线程在调用，那么就必须提供一个安全的发送函数
        auto iter = DynamicBidBuffers_.find(queueId);
        if (unlikely(iter == DynamicBidBuffers_.end())) {
            return 1;
        }
        auto ptr = iter->second->QueueWrapper;
        TRACE_DEBUG << ("another Thread send message:") << id << " to queue id:" << ptr->id();
        std::lock_guard<std::mutex> lock(iter->second->mutex_);
        return sendMsgInThread(ptr, buf, len, id);
    }

    int QueueManager::pubThreadData(const char* buf, uint32_t len, message_id_t id, thread_id_t tid)
    {
        auto iter = msgCodePubQueues_.find(id);
        if (unlikely(iter == msgCodePubQueues_.end())) {
            return -1;
        }
        /// 获取这个线程需要向哪些buffer发数据
        for (const auto& ptr : iter->second) {
            sendMsgInThread(ptr, buf, len, id);
            TRACE_DEBUG << " business Thread:" << tid << " send message:" << id << " to queue id:" << ptr->id();
        }
        return 0;
    }

    int QueueManager::pubThreadData(const char* buf, uint32_t len, message_id_t id, thread_id_t tid, queue_id_t bid)
    {
        auto iter = pubidQueues_.find(bid);
        if (unlikely(iter == pubidQueues_.end())) {
            return -1;
        }
        for (const auto& ptr : iter->second) {
            sendMsgInThread(ptr, buf, len, id);
            TRACE_DEBUG << " business Thread:" << tid << " send message:" << id << " to queue id:" << ptr->id();
        }
        return 0;
    }

    /// IPC线程发送消息到 业务线程
    int QueueManager::pubIPCDataToBusiThread(const char*     buf,
                                             uint32_t        len,
                                             message_id_t    id,
                                             thread_id_t     tid,
                                             connection_id_t connId)
    {
        /// 获取这个线程需要向哪些buffer发数据
        auto iter = ioThreadPubQueues_.find(tid);
        if (unlikely(iter == ioThreadPubQueues_.end())) {
            return -1;
        }
        /// TCP来的消息是不知道有哪些消息ID的，因此不需要进行msgcode的判断，msgcode只是为了线程间通信提前配置好的消息使用的
        for (const auto& ptr : iter->second) {
            sendMsgInThread(ptr, buf, len, id, connId, tid);
            TRACE_DEBUG << " IPC Thread:"_s << tid << " send message:"_s << id << " to queue id:"_s << ptr->id();
        }
        return 0;
    }

    std::vector<FIFOQueue::s_ptr> QueueManager::getBusiThreadSubQueues(thread_id_t tid)
    {
        std::vector<FIFOQueue::s_ptr> vec1;
        std::vector<FIFOQueue::s_ptr> vec2;
        auto                          iter1 = busiThreadSubTcpQueues_.find(tid);
        if (iter1 != busiThreadSubTcpQueues_.end()) {
            vec1 = iter1->second;
        }
        auto iter2 = busiThreadSubBusiQueues_.find(tid);
        if (iter2 != busiThreadSubBusiQueues_.end()) {
            vec2 = iter2->second;
        }
        vec2.insert(vec2.end(), vec1.begin(), vec1.end());
        return vec2;
    }

    const std::vector<queue_id_t>& QueueManager::GetPubIds(thread_id_t tid) const { return threadPubIds_.at(tid); }

}  // namespace foundation
}  // namespace xncc